package io.gitee.redismq.core.service;

import io.gitee.redismq.api.StreamTemplate;
import io.gitee.redismq.comm.constant.Constant;
import io.gitee.redismq.comm.exception.RedisMQException;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;

/**
 * @Description: RedisMQ服务抽象类
 * @Author: 孙浩
 * @CreateTime: 2025-03-06 09:46:31
 */
public class DefaultRedisMQTemplate implements StreamTemplate {

    protected final Executor pool;

    protected final RedisTemplate<String, Object> redisTemplate;

    protected DefaultRedisMQTemplate(Executor pool, RedisTemplate<String, Object> redisTemplate) {
        this.pool = pool;
        this.redisTemplate = redisTemplate;
    }


    @Override
    public <T> void sendMessage(final String streamKey, final T payload) {
        Map<String, Object> message = new HashMap<>();
        sendMessage(streamKey, payload, message);
    }


    @Override
    public <T> void sendMessage(String streamKey, T payload, Map<String, Object> headers){
        Map<String, Object> message = new HashMap<>(headers);
        message.put(Constant.DATA, payload);  // 统一字段名
        RecordId recordId = redisTemplate.opsForStream().add(streamKey, message);
        if (recordId == null) {
            throw new RedisMQException("Failed to send message to Redis Stream");
        }
    }
}
